package com.csust.common.hdfs;

import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.net.URI;
import java.net.URISyntaxException;
import java.util.ArrayList;
import java.util.List;

import com.csust.common.pojo.SystemConstants;
import lombok.extern.slf4j.Slf4j;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.*;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.io.IOUtils;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.security.UserGroupInformation;
import org.apache.hadoop.util.LineReader;
import static com.csust.common.pojo.SystemConstants.*;

/***
 * 通过kerberos认证的 hdfs operator
 * @author yhm
 * @date 2021/10/30
 * @descripe 因为用户权限问题，我将fileSystem封装成hdfs operator，避免出现目标文件夹权限不足的问题。
 */
@Slf4j
public class HdfsOp {
    // hdfs
    private static FileSystem fs = null;
    // kerberos认证信息
    private String userPrincipal = null;
    private String path2UserKeytab = null;
    // namenode的rpc地址
    private String namenodeRpcAddress = null;
    private String rootDir = null;



    public HdfsOp(String userPrincipal, String path2UserKeytab, String namenodeRpcAddress, String rootDir) {
        try {
            this.userPrincipal = userPrincipal;
            this.path2UserKeytab = path2UserKeytab;
            this.namenodeRpcAddress = namenodeRpcAddress;
            this.rootDir = rootDir;
            // 初始化文件系统
            System.setProperty(JVM_KRB5_CONF_PARM, JVM_KRB5_CONF_PARM_VALUE);
            Configuration conf = new Configuration();
            conf.set("fs.hdfs.impl", org.apache.hadoop.hdfs.DistributedFileSystem.class.getName());
            UserGroupInformation.setConfiguration(conf);
            UserGroupInformation.loginUserFromKeytab(userPrincipal, path2UserKeytab);
            fs = FileSystem.get(new URI(namenodeRpcAddress), conf);
        } catch (IOException e) {
            e.printStackTrace();
        } catch (URISyntaxException e) {
            e.printStackTrace();
        }
    }

    /**
     * 获取文件系统
     */
    public FileSystem getFileSystem() {
        return fs;
    }
    /**
     * 文件或文件夹是否存在
     * @param path 文件或文件夹地址
     * @return 是否存在
     */
    public boolean exist(String path) throws IOException {
        path = rootDir + path;
        return fs.exists(new Path(path));
    }

    /**
     * 新建文件夹
     * @param path 文件夹地址
     * @return 是否成功
     */
    public boolean mkdir(String path) throws IOException {
        path = rootDir + path;
        Path dirPath = new Path(path);
        if (!fs.exists(dirPath)) {
            fs.mkdirs(dirPath);
            log.info("success to create path " + path);
            return true;
        } else {
            log.warn(path + " already exist");
            return false;
        }
    }
    /**
     * 创建一个新的文件并写入数据
     * @param path 文件地址
     * @param content 一行字符
     */
    public void write(String path, String content) throws IOException {
        path = rootDir + path;
        FSDataOutputStream writeOut = null;
        try {
            writeOut = fs.create(new Path(path));
            writeOut.write(content.getBytes());
            writeOut.hsync();
            log.info("success to write " + path + " with String");
        } finally {
            IOUtils.closeStream(writeOut);
        }
    }
    /**
     * 创建一个新的文件并写入数据
     * @param path 文件地址
     * @param contents 多行字符
     */
    public void write(String path, String[] contents) throws IOException {
        path = rootDir + path;
        FSDataOutputStream writeOut = null;
        try {
            writeOut = fs.create(new Path(path));
            for (String content : contents) {
                writeOut.write(content.getBytes());
            }
            writeOut.hsync();
            log.info("success to write " + path + " with String[]");
        } finally {
            IOUtils.closeStream(writeOut);
        }
    }
    /**
     * 创建一个新的文件并写入数据
     * @param path 文件地址
     * @param contents 多行字符
     */
    public void write(String path, List<String> contents) throws IOException {
        path = rootDir + path;
        FSDataOutputStream writeOut = null;
        try {
            writeOut = fs.create(new Path(path));
            for (String content : contents) {
                writeOut.write(content.getBytes());
            }
            writeOut.hsync();
            log.info("success to write " + path + " with List<String>");
        } finally {
            IOUtils.closeStream(writeOut);
        }
    }
    /**
     * 文件尾部追加数据
     * @param path 文件地址
     * @param content 一行字符
     */
    public void append(String path, String content) throws IOException {
        path = rootDir + path;
        FSDataOutputStream appendOut = null;
        try {
            appendOut = fs.append(new Path(path));
            appendOut.write(content.getBytes());
            appendOut.hsync();
            log.info("success to append " + path + " with String");
        } finally {
            IOUtils.closeStream(appendOut);
        }
    }
    /**
     * 文件尾部追加数据
     * @param path 文件地址
     * @param contents 多行字符
     */
    public void append(String path, String[] contents) throws IOException {
        path = rootDir + path;
        FSDataOutputStream appendOut = null;
        try {
            appendOut = fs.append(new Path(path));
            for (String content : contents) {
                appendOut.write(content.getBytes());
            }
            appendOut.hsync();
            log.info("success to append " + path + " with String[]");
        } finally {
            IOUtils.closeStream(appendOut);
        }
    }


    /**
     * 文件尾部追加数据
     * @param path 文件地址
     * @param contents 多行字符
     */
    public void append(String path, List<String> contents) throws IOException {
        path = rootDir + path;
        FSDataOutputStream appendOut = null;
        try {
            appendOut = fs.append(new Path(path));
            for (String content : contents) {
                appendOut.write(content.getBytes());
            }
            appendOut.hsync();
            log.info("success to append " + path + " with List<String>");
        } finally {
            IOUtils.closeStream(appendOut);
        }
    }

    /**
     * 读取文件数据
     * @param path 文件地址
     * @return 每行数据
     */
    public List<String> read(String path) {
        path = rootDir + path;
        Path readPath = new Path(path);
        List<String> list = new ArrayList<String>();
        FSDataInputStream in = null;
        BufferedReader reader = null;
        try {
            in = fs.open(readPath);
            reader = new BufferedReader(new InputStreamReader(in));
            String sTempOneLine;
            while ((sTempOneLine = reader.readLine()) != null) {
                list.add(sTempOneLine);
            }
            log.info("success to read " + path);
        } catch (IOException e) {
            log.error("fail to read " + path + "\n" +
                    "ErrorMessage:" + e.getMessage());
        } finally {
            IOUtils.closeStream(reader);
            IOUtils.closeStream(in);
        }
        return list;
    }
    /**
     * 删除文件
     * @param path 文件地址
     * @return 是否成功
     */
    public boolean delFile(String path) throws IOException {
        path = rootDir + path;
        Path beDeletedPath = new Path(path);
        if (fs.delete(beDeletedPath, true)) {
            log.info("success to delete the file " + path);
            return true;
        } else {
            log.warn("failed to delete the file " + path);
            return false;
        }
    }

    /**
     * 删除文件夹
     * @param path 文件夹地址
     * @return 是否成功
     */
    public boolean delDir(String path) throws IOException {
        path = rootDir + path;
        Path delRecuDestPath = new Path(path);
        if (!fs.exists(delRecuDestPath)) {
            log.warn("dir " + path + " doesn't exist!");
            return false;
        }
        if (!fs.delete(delRecuDestPath, true)) {
            log.warn("fail to delete dir " + path);
            return false;
        }
        log.info("success to delete dir " + path);
        return true;
    }


    /**
     * 获取该用户根文件夹下指定目录下所有文件
     * @param path hdfs远端目录路径
     * @return 当前目录所有文件信息
     */
    public List<String> ls(String path) throws Exception {
        path = rootDir + path;
        String separator = "/";
        List<String> list = new ArrayList<String>();
        FileStatus[] statuses = fs.listStatus(new Path(path));
        for (FileStatus file : statuses) {
            String isDir = file.isDirectory() ? "dir" : "file";
            String permission = file.getPermission().toString();
            short replication = file.getReplication();
            long length = file.getLen();
            String fileOrDirPath = file.getPath().toString();
            list.add(isDir + separator
                    + permission + separator
                    + replication + separator
                    + length + separator
                    + fileOrDirPath);
        }
        return list;
    }
    /**
     * 上传资源到hdfs
     * @param src 本地文件地址
     * @param dst hdfs中用户文件夹的相对地址
     * @return 是否上传成功
     */
    public boolean put(String src, String dst) {
        try {
            dst = rootDir + dst;
            fs.copyFromLocalFile(false, new Path(src), new Path(dst));
            log.info("src(local):" + src + "=>dst(hdfs):" + dst);
            return true;
        } catch (Exception e) {
            log.error(e.getMessage());
            return false;
        }
    }

    /**
     * 从hdfs下载资源到本地
     * @param src hdfs中用户文件夹的相对地址
     * @param dst 本地文件地址
     * @return 是否下载成功
     */
    public boolean get(String src, String dst) {
        try {
            src = rootDir + src;
            fs.copyToLocalFile(false, new Path(src), new Path(dst), true);
            log.info("src(hdfs):" + src + "=>dst(local):" + dst);
            return true;
        } catch (Exception e) {
            log.error(e.getMessage());
            return false;
        }

    }

    /**
     * 释放FileSystem资源
     */
    public void close() throws IOException {
        fs.close();
    }

    /**
     * 当对象被GC的时候先释放资源，尽量手动释放资源，减少full GC
     */
    @Override
    protected void finalize() throws Throwable {
        super.finalize();
        fs.close();
    }

    public static void main(String[] args) throws Exception {
        HdfsOp hdfsOp = HdfsOpFactory.getHdfsOp(SystemConstants.USER3_PRINCIPAL);
        hdfsOp.mkdir("/test");
        hdfsOp.write("/HelloHDFS.txt", "HelloHDFS");
        hdfsOp.put("./common/src/main/resources/krb5.conf","/krb5.conf");
        hdfsOp.get("/krb5.conf","./krb5.conf");
        System.out.println(hdfsOp.read("/HelloHDFS.txt"));
        System.out.println(hdfsOp.ls("/"));
        hdfsOp.close();
    }


}